<?php
/*
    Lightweight and multiplatform PHP Multithreading Engine
    Copyright (C) 2008  Anton Vedeshin (anton@vedeshin.com, http://anton.vedeshin.com)
	
    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

error_reporting(E_ALL);

//### PARAMETERS
$threadnum=30; //number of threads to be executed

//### HELPING FUNCTIONS
//logger, sleeper, SQLescape
require_once("functions.php");

//### THREADS
//Class Thread, main class that does threading
require_once ("Thread.php");

//### DATABASE SETTINGS
define ("s_host","yourdbhost");
define ("s_login","yourlogin");
define ("s_password","yourpassword");
define ("s_db_name","yourdbname");
define ("s_db_name_threads","threads"); //dbname for threads, it could be different

//### DATABASE INCLUDES AND CONNECTION
// Include ezSQL core
include_once "x:/your_path_to_ezSQL/db/shared/ez_sql_core.php";
// Include ezSQL database specific component
include_once "x:/your_path_to_ezSQL/db/mssql/ez_sql_mssql.php";

$db=new ezSQL_mssql(s_login, s_password, s_db_name, s_host); //connection for your application, i/o of parameters and results etc.
$db2=new ezSQL_mssql(s_login, s_password, s_db_name_threads, s_host); //additional connection for database used in threading only, could be the same

//### Logger settings and initialization
logger::deleteold(); //delete old logs
$l=new logger("main");
$l->dolog(true);
$l->lw(1,"-----------------------------------------------------------\n");
$l->lw(1,"All is included and prepared to work");

//### GET PARAMETERS AND INPUT DATA FOR WORK INSIDE THREADS
$result=$db->get_results("SELECT field1, field2, field3, ... FROM input_table");

if(!$result){
	echo count($result);
	echo "there is no job!!!";
	exit();
}

//### THREADS
//### cleaning
$rowsaffected=$db2->query("DELETE FROM threads");
$l->lw(1,"DB threads are cleaned, rows affected: ".$rowsaffected);
$rowsaffected=$db2->query("DELETE FROM cmd");
$l->lw(1,"DB cmd are cleaned, rows affected: ".$rowsaffected);



//### CREATE THREADS
$threads=array();
for($i=1;$i<=$threadnum;$i++){
	$proc_id=7000+$i;
	$threads[$proc_id] = Thread::create("import_ignitor_thread.php",$proc_id);
	$l->lw(1,"Thread [".$proc_id."] created.");
}

$l->lw(1,"All threads are created.");

//### prepare work for THREADS 
$jobs=array();
$work_volume=count($result);
$chunksize=intval($work_volume/$threadnum); //$chunknum
$additional=$work_volume-$chunksize*$threadnum;

$l->lw(4,"work volume: $work_volume, chunk size: $chunksize, odd: $additional");

//### prepare $jobs object, put your own object with needed structure into $jobs
$add=0;
for($i=0;$i<$threadnum;$i++){
	if($additional>0){
		$additional--;
		$add=1;
	}
	for ($j=$i*$chunksize;$j<$i*$chunksize+$chunksize+$add;$j++){
		 $jobs[$i][]=$result[$j]->field1."";
	}
	$add=0;
}

//### check that threads are created and ready
$l->lw(1,"Check that all threads are created and ready.");
$resultx=$db2->get_var("SELECT count(*) FROM threads WHERE state=1");
if($resultx==0){
  $l->lw(1,"Not all threads were created. Created ".$result." of ".$threadnum);
  die("Error! Problem with execution of threads, application terminated");
}	

//### THREADS main cycle
//BEAWARE OF TIMINGS (Sleeper), it could affects the perfomanse or collisions could occur
do{
	$q = true;
	$closed_threads=0;
	foreach ($threads as $proc_id => $thread) {
		if ($thread->isActive()){//if is not busy
			if (!$thread->isBusy()){//free, put work
				$l->lw(1,"Is not busy nr.: ".$thread->proc_id);
				if(count($jobs)>0){//if work exists, load
					$thread->tell("calculate",array_pop($jobs));
					$l->lw(1,"Jobs left: ".count($jobs));
				}else{
					$l->lw(1,"Terminating thread nr.: ".$thread->proc_id);
					$thread->tell("exit");
				}
				//Sleeper(300);
			}else{//busy, do not touch
				Sleeper(10000);
			}
		}else{
			$closed_threads++;
			//future development:
			//if is not responding/busy
			//try to exit thread
			//wait 5 sec
			//kill
		}
	}
	if($closed_threads==$threadnum){
		$l->lw(1,"All threads are closed, closing...");
		foreach ($threads as $proc_id => $thread) {
			$thread->tell("exit");
		}
		$q=false;
		break;
	}
	Sleeper(30000);
}while($q);

//### Get results
$results=$db2->get_results("SELECT result FROM cmd WHERE done=1");

if(count($results)<$threadnum){
	$l->lw(1,"Not all results returned, num: ".count($results));
	$results2=$db2->get_results("SELECT proc_id FROM cmd WHERE done=0 and cmd LIKE 'calculate'");
	$l->lw(1,"Did not returned, num: ".count($results2));
	$procs="";
	if(count($results2)>0){
	  foreach ($results2 as $result2){
		 $procs.=$result2->proc_id.", ";
	  }
	}
	$l->lw(1,"Did not returned threads: ".$procs);
}else {
	$l->lw(1,"All the results were returned");
}

//### Store the results
// you can also store results in file etc.
$db3=new ezSQL_mssql("your_login", "your_password", "your_db_name", "your_host");
	$bquery="BEGIN TRANSACTION \r\n";
	$equery="COMMIT TRANSACTION";
	
	if(count($results)>0){
		foreach ($results as $result){
			$queryX="";
			$array=unserialize(base64_decode($result->result));
			$queryX=$bquery.$array[0]."\r\n".$equery;
			$db3->query($queryX);
		}
	}

$l->lw(1,"Exited normally!");

?>

